/********************************************************/
/*	Author:		gong_libin			*/
/*	Date:		2017_11_01			*/
/*	File:		CmAdxEng.cpp			*/
/********************************************************/

#include "CmAdxEng.h"

static CCmConfig	sg_CConf;
static CCmRedis*	sg_pCRds;
static CCmProducer*	sg_pCProd;

int main(int argc, char* argv[])
{
 	int iIndex = 0;
	int iReturn = 0;
	int iThread = 1;
	CmCls::CCmThread* pCThread = NULL;
	char* pszConf = (char*)"../../conf/CmAdxEng.conf";

	struct option stOptions[] = {
		{"config", 1, 0, 'c'},
		{"thread", 1, 0, 't'},
		{"version", 0, 0, 'v'},
		{"help", 0, 0, 'h'},
		{0, 0, 0, 0},
	};

	while (CM_FAILURE != (iReturn = getopt_long(argc, argv, "c:t:vh", stOptions, &iIndex))) {
		switch (iReturn) {
			case 'c':
				pszConf = optarg;
				break;
			case 't':
				iThread = atoi(optarg);
				break;
			case 'v':
				CM_PRINT("Copyright (C)2017 Charm Media Group\n");
				CM_PRINT("%s Build on %s %s.\n", CM_ADX_VERS, __DATE__, __TIME__);
				return CM_SUCCESS;
			case 'h':
				CM_PRINT("Usage: %s [options]\n", argv[0]);
				CM_PRINT("Valid options are:\n");
				CM_PRINT("[0] -c or --config:     set config\n");
				CM_PRINT("[0] -t or --thread:     set thread\n");
				CM_PRINT("[0] -v or --version:    show version\n");
				CM_PRINT("[0] -h or --help:       show help\n");
				CM_PRINT("Copyright (C)2017 Charm Media Group\n");
				CM_PRINT("%s Build on %s %s.\n", CM_ADX_VERS, __DATE__, __TIME__);
				return CM_SUCCESS;
			default:
				return CM_SUCCESS;
		}
	}

	signal(SIGINT, CmAdxEngQuit);
	signal(SIGSEGV, CmAdxEngQuit);
	signal(SIGTERM, CmAdxEngQuit);

	if (CM_SUCCESS == sg_CConf.CmConfigLoad(pszConf)) {
		#ifdef _CM_DEBUG_
			iThread = 8;
		#else
			CmFork();
		#endif
		sg_CConf.CmConfigPrint();
		if (CM_SUCCESS == CCmEvent::CmEventInit((sg_CConf.CmConfigGetGlobal())->m_szPort)) {
			CCmEvent::CmEventPthread();
			pCThread = new CmCls::CCmThread[iThread];
			for (iIndex = 0; iIndex < iThread; iIndex ++) {
				(pCThread + iIndex)->CmThreadAttrInit();
				(pCThread + iIndex)->CmThreadRoutineSet(CmAdxEngThread);
				(pCThread + iIndex)->CmThreadCreate((void*)new CCmEvent());
			}

			sg_pCRds = new CCmRedis(**(CMSTR_C**)sg_CConf.CmConfigGetRedis()->CmSetIndexGet(0));
			sg_pCProd = new CCmProducer(sg_CConf.CmConfigGetKafka()->m_szBroke, sg_CConf.CmConfigGetKafka()->m_szTopic);

			CCmAdmin CAdmin(*sg_CConf.CmConfigGetCTrans(), sg_CConf.CmConfigGetPTrans());
			CAdmin.CmThreadRoutineSet(CmCls::CmThreadType<CCmAdmin>);
			CAdmin.CmThreadAttrInit();
			CAdmin.CmThreadCreate(&CAdmin);
			CAdmin.CmThreadJoin(NULL);
		}
		else {
			CM_ERROR("Failed to CmEventInit\n");
		}
	}
	else {
		CM_ERROR("Failed to CmConfigLoad\n");
	}

	return CM_SUCCESS;
}

void CmAdxEngQuit(int iSignal)
{
	CM_ERROR("%s stopped by signal %d\n", CM_ADX_VERS, iSignal);

	exit(CM_FAILURE);
}

void* CmAdxEngThread(void* pPara)
{
	CCmEvent* pCEvent = (CCmEvent*)pPara;
	pCEvent->CmEventCall(CmAdxEngCore, NULL);

	return pCEvent->CmEventDispatch();
}

void CmAdxEngCore(struct evhttp_request* pstReq, void* pNULL)
{
	CMADXPARA_S* pstPara = NULL;

	#ifdef _CM_DEBUG_
		CmCls::CCmTime CTime;
		CTime.CmTimeOn();
		CM_PRINT("\nCmAdxEngCore[%lu] start\n", pthread_self());
	#endif

	if (NULL != (pstPara = CmAdxEngRequest(pstReq))) {
		pstPara->m_pCAdxLog = new CMMAP_C();
		if (CM_SUCCESS == CmAdxEngDistribute(pstPara)) {
			#ifdef _CM_DEBUG_
				CM_DEBUG("CmAdxEngDistribute: success\n");
			#endif
		}
		else {
			#ifdef _CM_DEBUG_
				CM_DEBUG("CmAdxEngDistribute: failure\n");
			#endif
		}
		CmAdxEngResponse(pstReq, pstPara);
		CmAdxEngWriteLog(pstPara);
	}

	#ifdef _CM_DEBUG_
		CM_PRINT("CmAdxEngCore[%lu] end Use time ", pthread_self());
		CTime.CmTimeOff();
		CTime.CmTimeDisplay();
	#endif

	return;
}

int CmAdxEngDistribute(CMADXPARA_S* pstPara)
{
	CMVEC_C CVec;
	CCmProJsn CPro;
	CMSTR_C CDsp, CLog;
	CMLINK_S* pstLink = NULL;
	CMMARK_S* pstMark = NULL;
	CMMARK_S stMark = {0, NULL};
	CCmTrans** ppstTrans = NULL;
	CmCls::CCmSet CTrans, COffer;
	CMADXMSG_S stTemp = {0, {'\0'}};
	CMFEED_S stFeed = {0, {0, {'\0'}}};
	int iMax = 0, iReturn = CM_SUCCESS;
	ULONG ulIndex = 0, ulTrans = 0xffff;
	fd_set iTemp = {0}, iWait = {0}, iZero = {0};
	struct timeval stTim = {0, 0}, stOut = {0, 0};

	CPro.CmProtoSerialize(pstPara);
	sprintf((char*)stTemp.m_uszMsg, CM_ADX_POST, pstPara->m_stDspReq.m_uiMsg);
	sprintf((char*)stTemp.m_uszMsg + (stTemp.m_uiMsg = strlen((char*)stTemp.m_uszMsg)), "%s", pstPara->m_stDspReq.m_uszMsg);

	COffer.CmSetInit((*sg_CConf.CmConfigGetCTrans()).CmMapSize(), sizeof(CMFEED_S));
	CTrans.CmSetInit((*sg_CConf.CmConfigGetCTrans()).CmMapSize(), sizeof(CMMARK_S));

	#ifdef _CM_DEBUG_
		CM_DEBUG("Insert CTrans: ");
	#endif
	//sg_pCRds->smembers(CMSTR_C(CM_ADX_DSPS), &CVec);
	CVec.push_back("1");
	CVec.push_back("2");
	CVec.push_back("3");
	CVec.push_back("4");
	CVec.push_back("5");
	for (ulIndex = 0; ulIndex < CVec.size(); ++ ulIndex) {
		if ((ulTrans = CmAdxEngGetTrans(CVec[ulIndex])) < CVec.size()) {
			ppstTrans = *sg_CConf.CmConfigGetPTrans() + ulTrans;
			#ifdef _CM_DEBUG_
				CM_PRINT("%s -> ", (*ppstTrans)->CmTransName());
			#endif
			if (CM_LNK_NUM != (stMark.m_uiCursor = (*ppstTrans)->CmTransGetOne())) {
				pstLink = (*ppstTrans)->CmTransLink() + stMark.m_uiCursor;
				pstLink->m_CTcp.CmNetSend(stTemp.m_uszMsg, stTemp.m_uiMsg + pstPara->m_stDspReq.m_uiMsg);
				iMax = iMax < pstLink->m_CTcp.CmNetGetSocket() ? pstLink->m_CTcp.CmNetGetSocket() : iMax;
				#ifdef _CM_DEBUG_
					CM_PRINT("%d, ", pstLink->m_CTcp.CmNetGetSocket());
				#endif
				stMark.m_ppCTrans = ppstTrans;
				FD_SET(pstLink->m_CTcp.CmNetGetSocket(), &iTemp);
				CTrans.CmSetIndexPut(CTrans.CmSetGetCount(), &stMark);
			}
			else {
				#ifdef _CM_DEBUG_
					CM_PRINT("NULL, ");
				#endif
			}
		}
		else {
			#ifdef _CM_DEBUG_
				CM_PRINT("%s error, ", CVec[ulIndex].c_str());
			#endif
		}
	}
	#ifdef _CM_DEBUG_
		CM_PRINT("\n");
	#endif

	if (iMax > 0) {
		while (CM_SUCCESS == CmAdxEngSetTimOut(&stTim, &stOut) && CM_SUCCESS != memcmp(&iTemp, &iZero, sizeof(fd_set))) {
			#ifdef _CM_DEBUG_
				CM_DEBUG("Will wait %ld second %ld microsecond\n", stTim.tv_sec, stTim.tv_usec);
			#endif
			memcpy(&iWait, &iTemp, sizeof(fd_set));
			switch(select(iMax + 1, &iWait, NULL, NULL, &stTim)) {
				case CM_SUCCESS:
					#ifdef _CM_DEBUG_
						CM_DEBUG("Select timeout!!!\n");
					#endif
					break;
				case CM_FAILURE:
					#ifdef _CM_DEBUG_
						CM_DEBUG("Select error!!!\n");
					#endif
					break;
				default:
					for (ulIndex = 0; ulIndex < CTrans.CmSetGetCount(); ++ ulIndex) {
						pstMark = (CMMARK_S*)CTrans.CmSetIndexGet(ulIndex);
						CDsp = CMSTR_C((*pstMark->m_ppCTrans)->CmTransName());
						pstLink = (*pstMark->m_ppCTrans)->CmTransLink() + pstMark->m_uiCursor;
						if (0 != FD_ISSET(pstLink->m_CTcp.CmNetGetSocket(), &iWait)) {
							#ifdef _CM_DEBUG_
								CM_DEBUG("%d event: ", pstLink->m_CTcp.CmNetGetSocket());
							#endif
							memset(&stFeed, '\0', sizeof(CMFEED_S));
							FD_CLR(pstLink->m_CTcp.CmNetGetSocket(), &iTemp);
							if ((stTemp.m_uiMsg = pstLink->m_CTcp.CmNetRecv(stTemp.m_uszMsg, CM_KBYTES8)) > 0) {
								#ifdef _CM_DEBUG_
									CM_PRINT("read %d bytes ", stTemp.m_uiMsg);
								#endif
								(*pstMark->m_ppCTrans)->CmTransPayOne(pstLink);
								if (0 != CmAdxEngHttpParse(&stFeed.m_stMsg, &stTemp)) {
									#ifdef _CM_DEBUG_
										CM_PRINT("and parse success\n");
									#endif
									stFeed.m_uiDsp = atoi((char*)CDsp.c_str());
									COffer.CmSetIndexPut(COffer.CmSetGetCount(), &stFeed);
									CLog = CMSTR_C((char*)stFeed.m_stMsg.m_uszMsg);
								}
								else {
									#ifdef _CM_DEBUG_
										CM_PRINT("and parse failure\n");
									#endif
									CLog = CMSTR_C(CM_ADX_NULL);
								}
							}
							else {
								#ifdef _CM_DEBUG_
									CM_PRINT("be closed\n");
								#endif
								(*pstMark->m_ppCTrans)->CmTransCloseOne(pstLink);
								CLog = CMSTR_C(CM_ADX_NULL);
							}
							pstPara->m_pCAdxLog->CmMapInsert(CDsp, CLog);
						}
					}
					break;
			}
		}

		if (CM_SUCCESS != memcmp(&iTemp, &iZero, sizeof(fd_set))) {
			#ifdef _CM_DEBUG_
				CM_DEBUG("Destroy CTrans: ");
			#endif
			for (ulIndex = 0; ulIndex < CTrans.CmSetGetCount(); ++ ulIndex) {
				pstMark = (CMMARK_S*)CTrans.CmSetIndexGet(ulIndex);
				CDsp = CMSTR_C((*pstMark->m_ppCTrans)->CmTransName());
				pstLink = (*pstMark->m_ppCTrans)->CmTransLink() + pstMark->m_uiCursor;
				if (0 != FD_ISSET(pstLink->m_CTcp.CmNetGetSocket(), &iTemp)) {
					CLog = CMSTR_C(CM_ADX_NULL);
					pstPara->m_pCAdxLog->CmMapInsert(CDsp, CLog);
					(*pstMark->m_ppCTrans)->CmTransTimeout(pstLink);
					#ifdef _CM_DEBUG_
						CM_PRINT("%s -> %d", CDsp.c_str(), pstLink->m_CTcp.CmNetGetSocket());
					#endif
				}
			}
			#ifdef _CM_DEBUG_
				CM_PRINT("\n");
			#endif
		}

		if (COffer.CmSetGetCount() > 0) {
			#ifdef _CM_DEBUG_
				CM_DEBUG("Will CmAdxEngBidPrice ...\n");
			#endif
			CmAdxEngBidPrice(COffer, pstPara);
			CPro.CmProtoDeserialize(pstPara);
		}
		else {
			#ifdef _CM_DEBUG_
				CM_DEBUG("Skip CmAdxEngBidPrice !!!\n");
			#endif
		}
	}
	else {
		iReturn = CM_FAILURE;
	}

	CTrans.CmSetDestroy();
	COffer.CmSetDestroy();

	return iReturn;
}

void CmAdxEngBidPrice(CmCls::CCmSet& rCOffer, CMADXPARA_S* pstPara)
{
	CCmProJsn CJsn;
	ULONG ulWin = 0;
	UINT uiPrice = 0;
	CMADXINFO_S stWin;
	CMADXINFO_S* pstInfo = NULL;

	for (ULONG ulIdx = 0; ulIdx < rCOffer.CmSetGetCount(); ++ ulIdx) {
		if (NULL != (pstInfo = CJsn.CmProtoValidity((char*)((CMFEED_S*)rCOffer.CmSetIndexGet(ulIdx))->m_stMsg.m_uszMsg))) {
			#ifdef _CM_DEBUG_
				CM_DEBUG("DSP[%ld] is valid ", ulIdx);
			#endif
			if (stWin.m_uiMaxPrice > 0) {
				if (pstInfo->m_uiMaxPrice > stWin.m_uiMaxPrice) {
					#ifdef _CM_DEBUG_
						CM_PRINT("and price is %d > max price %d\n", pstInfo->m_uiMaxPrice, stWin.m_uiMaxPrice);
					#endif
					uiPrice = stWin.m_uiMaxPrice + 1;
					stWin = pstInfo;
					ulWin = ulIdx;
				}
				else {
					#ifdef _CM_DEBUG_
						CM_PRINT("and price is %d : max price %d\n", pstInfo->m_uiMaxPrice, stWin.m_uiMaxPrice);
					#endif
					uiPrice = uiPrice < pstInfo->m_uiMaxPrice ? pstInfo->m_uiMaxPrice + 1 : uiPrice;
				}
			}
			else {
				#ifdef _CM_DEBUG_
					CM_PRINT("and price is %d\n", pstInfo->m_uiMaxPrice);
				#endif
				stWin = pstInfo;
				ulWin = ulIdx;
			}
		}
		else {
			#ifdef _CM_DEBUG_
				CM_DEBUG("DSP[%ld] is invalid\n", ulIdx);
			#endif
		}
	}

	if (stWin.m_uiMaxPrice > 0) {
		stWin.m_uiCode = CM_ADF_HBID;
		stWin.m_uiMethod = CM_ADF_SCPM;
		stWin.m_uiMaxPrice = uiPrice > 0 ? uiPrice : stWin.m_uiMaxPrice;
		stWin.m_uiDsp = ((CMFEED_S*)rCOffer.CmSetIndexGet(ulWin))->m_uiDsp;

		CMSTR_C CData = CJsn.CmProtoSerialize(&stWin);
		memcpy(&pstPara->m_stDspRes.m_uszMsg, CData.c_str(), pstPara->m_stDspRes.m_uiMsg = CData.length());
		#ifdef _CM_DEBUG_
			CM_DEBUG("Final -> DSP[%ld] is winner\n", ulWin);
		#endif
	}

	return;
}

CMADXPARA_S* CmAdxEngRequest(struct evhttp_request* pstReq)
{
	CCmEvent CEvent;
	CMADXPARA_S* pstPara = NULL;

	#ifdef _CM_DEBUG_
		CMMAP_C* pCKvs = NULL;
		if (NULL != (pCKvs = CEvent.CmEventKvsGet(pstReq->uri_elems))) {
			if (true != pCKvs->CmMapEmpty()) {
				pCKvs->CmMapDisplay();
			}
			else {
				CM_DEBUG("Para: NULL\n");
			}
			CEvent.CmEventKvsDel(pCKvs);
		}
		else {
			CM_DEBUG("Para: ERROR\n");
		}

		CM_DEBUG("Host: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Host"));
		CM_DEBUG("Accept: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Accept"));
		CM_DEBUG("Connection: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Connection"));
		CM_DEBUG("User-Agent: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "User-Agent"));
		CM_DEBUG("Content-Type: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Content-Type"));
		CM_DEBUG("Content-Length: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Content-Length"));
		CM_DEBUG("Accept-Encoding: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Accept-Encoding"));
		CM_DEBUG("Accept-Language: %s\n", CEvent.CmEventFindHeader(pstReq->input_headers, "Accept-Language"));
	#endif

	pstPara = CEvent.CmEventContent(pstReq->input_buffer);
	#ifdef _CM_DEBUG_
		CM_DEBUG("Content: %s\n", NULL != pstPara ? (char*)pstPara->m_stSspReq.m_uszMsg : "NULL");
	#endif

	return pstPara;
}

int CmAdxEngResponse(struct evhttp_request* pstReq, CMADXPARA_S* pstPara)
{
	CCmEvent CEvent;

	CEvent.CmEventAddHeader(pstReq->output_headers, (char*)"Connection", (char*)"keep-alive");
	CEvent.CmEventAddHeader(pstReq->output_headers, (char*)"Cache-Control", (char*)"no-cache");
	CEvent.CmEventAddHeader(pstReq->output_headers, (char*)"Content-Type", (char*)"application/json;charset=utf-8");

	CEvent.CmEventSendReply(pstReq, pstPara->m_stSspRes.m_uszMsg, pstPara->m_stSspRes.m_uiMsg);

	return CM_SUCCESS;
}

void CmAdxEngWriteLog(CMADXPARA_S* pstPara)
{
	CCmProJsn CProReq, CProRes;
	CmCls::CCmMap<CMSTR_C, CMSTR_C, CmCls::CMMAPSTRING_O>::CM_ITE CIte = pstPara->m_pCAdxLog->CmMapBegin();
	CMADXINFO_S *pstInfoRes = NULL, *pstInfoReq = CProReq.CmProtoDeserialize((char*)pstPara->m_stSspReq.m_uszMsg);

	while (CIte != pstPara->m_pCAdxLog->CmMapEnd()) {
		// dsp request
		CCmLogger CLogReq(pstInfoReq);
		CLogReq.CmLoggerSetMsgType(CM_ADF_QDSP);
		CLogReq.CmLoggerSetDspId(atoi((char*)CIte->first.c_str()));

		// dsp response
		CCmLogger* pCLogRes = NULL;
		if (NULL != (pstInfoRes = CProRes.CmProtoDeserialize((char*)CIte->second.c_str()))) {
			pCLogRes = new CCmLogger(pstInfoRes);
			if (pstInfoRes->m_uiMaxPrice > 0) {
				pCLogRes->CmLoggerSetResCode(CM_ADF_HBID);
			}
			else {
				pCLogRes->CmLoggerSetResCode(CM_ADF_RSUC);
			}
		}
		else {
			pCLogRes = new CCmLogger(pstInfoReq);
			pCLogRes->CmLoggerSetResCode(CM_ADF_RFAL);
		}
		pCLogRes->CmLoggerSetDspId(atoi((char*)CIte->first.c_str()));
		pCLogRes->CmLoggerSetMsgType(CM_ADF_PDSP);

		CMSTR_C CPktReq = CLogReq.CmLoggerSerialize();
		CMSTR_C CPktRes = pCLogRes->CmLoggerSerialize();

		#ifdef _CM_DEBUG_
			//CM_PRINT("Req: %s\n", CPktReq.c_str());
			//CM_PRINT("Res: %s\n", CPktRes.c_str());
		#endif
		//sg_pCProd->CmProducerSend(CM_KFK_PART, (char*)CPktReq.c_str(), CPktReq.length());
		//sg_pCProd->CmProducerSend(CM_KFK_PART, (char*)CPktRes.c_str(), CPktRes.length());

		CIte ++;
	}

	CCmEvent().CmEventParaFree(pstPara);

	return;
}

int CmAdxEngSetTimOut(struct timeval* pstTim, struct timeval* pstOut)
{
	struct timeval stNow;
	int iReturn = CM_SUCCESS;
	UINT uiUsec = atoi((sg_CConf.CmConfigGetGlobal())->m_szTmOut);

	pstTim->tv_sec = 0;
	gettimeofday(&stNow, NULL);

	if (0 != pstOut->tv_sec && 0 != pstOut->tv_usec) {
		if (stNow.tv_sec == pstOut->tv_sec) {
			if ((pstTim->tv_usec = pstOut->tv_usec - stNow.tv_usec) <= 0) {
				iReturn = CM_FAILURE;
			}
		}
		else if (stNow.tv_sec < pstOut->tv_sec) {
			pstTim->tv_usec = 1000000L - stNow.tv_usec + pstOut->tv_usec;
		}
		else {
			iReturn = CM_FAILURE;
		}
	}
	else {
		pstTim->tv_usec = uiUsec;
		if ((stNow.tv_usec += uiUsec) > 1000000L) {
			pstOut->tv_usec = (stNow.tv_usec -= 1000000L);
			pstOut->tv_sec = (stNow.tv_sec += 1);
		}
		else {
			pstOut->tv_usec = stNow.tv_usec;
			pstOut->tv_sec = stNow.tv_sec;
		}
	}

	return iReturn;
}

int CmAdxEngHttpParse(CMADXMSG_S* pstRecv, CMADXMSG_S* pstTemp)
{
	char* pszCur = NULL;
	char* pszHttp = (char*)pstTemp->m_uszMsg;
	int iNum = 0, iLen = strlen(CM_ADX_CTSZ);

	while (NULL != (pszCur = strsep(&pszHttp, CM_ADX_CRLF))) {
		if ('\0' != *pszCur) {
			iNum = 0;
			if (CM_SUCCESS == strncasecmp(pszCur, CM_ADX_CTSZ, iLen)) {
				pstRecv->m_uiMsg = atoi(pszCur + iLen);
			}
		}
		else {
			if (CM_ADX_RRNN == ++ iNum) {
				memcpy(pstRecv->m_uszMsg, pszCur + 1, pstRecv->m_uiMsg);
				break;
			}
		}
	}

	return pstRecv->m_uiMsg;
}

ULONG CmAdxEngGetTrans(CMSTR_C& rCId)
{
	ULONG ulIndex = 0;

	while (ulIndex < (*sg_CConf.CmConfigGetCTrans()).CmMapSize()) {
		if (CMSTR_C((*(*sg_CConf.CmConfigGetPTrans() + ulIndex))->CmTransName()) == rCId) break;
		++ ulIndex;
	}

	return ulIndex;
}
